package cn.wx.read.kafka.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import cn.hutool.core.lang.Singleton;
import cn.wx.common.persistence.base.BaseController;
import cn.wx.common.persistence.web.ResponseData;
import cn.wx.read.kafka.pool.ExecutorPool;
import cn.wx.read.kafka.pool.FilePool;
import cn.wx.read.kafka.pool.Gone;
import cn.wx.read.kafka.service.ConsumeBookService;
import cn.wx.read.kafka.service.ConsumeExpertService;
import cn.wx.read.kafka.service.ConsumePaperService;
import cn.wx.read.kafka.service.ConsumePatentsService;
import cn.wx.read.kafka.service.RelationEbService;
import cn.wx.read.kafka.service.RelationEfoService;
import cn.wx.read.kafka.service.RelationEorService;
import cn.wx.read.kafka.service.RelationEpService;
import cn.wx.read.kafka.service.RelationEptService;

@RestController
@RequestMapping("/consume")
public class ConsumeController extends BaseController {
	
	@Autowired
    private ConsumeExpertService consumeExpertService;
	
	@Autowired
    private ConsumeBookService consumeBookService;
	
	@Autowired
    private ConsumePaperService consumePaperService;
	
	@Autowired
    private ConsumePatentsService consumePatentsService;

	@Autowired
	private RelationEbService relationEbService;
	
	@Autowired
	private RelationEpService relationEpService;
	
	@Autowired
	private RelationEptService relationEptService;
	
	@Autowired
	private RelationEfoService relationEfoService;
	
	@Autowired
	private RelationEorService relationEorService;
	
	
	ExecutorPool executorPool = Singleton.get(ExecutorPool.class);
	
    /**
     * @author :  
     * @since : Create in 2018-11-13
     */
    @RequestMapping(value = "/consume1", method = RequestMethod.POST)
    public ResponseData consumeExpert(@RequestParam(value="path") String path) throws Exception {

    	long s_ = System.currentTimeMillis();
    	
    	/**
    	 * 插入专家
    	 */
    	consumeExpertService.consumeZone(path,FilePool.ms_expert_info,Gone.Neo4j,"ZoneExpert_");
    	executorPool.lock();
    	/**
    	 * 插入图书
    	 */
    	consumeBookService.consumeZone(path,FilePool.ms_expert_book,Gone.Neo4j);
    	executorPool.lock();

    	/**
    	 * 插入论文
    	 */
    	consumePaperService.consumeZone(path,FilePool.ms_expert_paper,Gone.Neo4j);
    	executorPool.lock();
    	
    	/**
    	 * 插入专利
    	 */
    	consumePatentsService.consumeZone(path,FilePool.ms_expert_patent,Gone.Neo4j);
    	executorPool.lock();
    	
    	/**
    	 * 专家和图书的关系
    	 */
    	relationEbService.consumeZone(path,FilePool.expertBookRelation);
    	executorPool.lock();
    	
    	/**
    	 * 专家和论文的关系
    	 */
    	relationEpService.consumeZone(path,FilePool.expertPaperRelation);
    	executorPool.lock();
    	
    	/**
    	 * 专家和专利的关系
    	 */
    	relationEptService.consumeZone(path,FilePool.expertPatentRelation);
    	executorPool.lock();

    	/**
    	 * 专家和领域的关系
    	 */
    	relationEfoService.consumeZone(path, FilePool.expertFieldRelation,"ZoneRelationEfo_");
    	executorPool.lock();
    	
    	
    	/**
    	 * 专家和领域的关系paperCount
    	 */
    	relationEfoService.consumeZone(path, FilePool.expertFieldRelation,"ZoneRelationEfoPaperCount_");
    	executorPool.lock();
    	
    	/**
    	 * 专家和领域的关系rank
    	 */
    	relationEfoService.consumeZone(path, FilePool.expertFieldRelation,"ZoneRelationEfoRank_");
    	executorPool.lock();
    	
    	
    	/**
    	 * 专家和合作机构的关系
    	 */
    	relationEorService.consumeZone(path, FilePool.expertOrgRelation);
    	executorPool.lock();
    	
    	
    	/**
    	 * --------------------------------以上插入专家到 neo4j--------------------------------------------------
    	 */
    	
    	/**
    	 * 插入专家到es
    	 */
    	consumeExpertService.consumeZone(path,FilePool.ms_expert_info,Gone.Elasticsearch,"ZoneExpert_");
    	executorPool.lock();
    	
    	/**
    	 * 插入专家ghindex到es
    	 */
    	consumeExpertService.consumeZone(path,FilePool.expert_ghindex,Gone.Elasticsearch,"ZoneExpertGHIndex_");
    	executorPool.lock();
    	
    	/**
    	 * 插入专家专利统计数据到es
    	 */
    	consumeExpertService.consumeZone(path,FilePool.industrial_patents,Gone.Elasticsearch,"ZoneExpertPatents_");
    	executorPool.lock();
    	
    	/**
    	 * 插入专家新闻社交统计数据到es
    	 */
    	consumeExpertService.consumeZone(path,FilePool.newsandsocial,Gone.Elasticsearch,"ZoneExpertNews_");
    	executorPool.lock();
    	
    	/**
    	 * --------------------------------以上插入专家到 Elasticsearch--------------------------------------------------
    	 */
    	
    	/**
    	 * 插入图书
    	 */
    	consumeBookService.consumeZone(path,FilePool.ms_expert_book,Gone.Mysql);
    	executorPool.lock();

    	/**
    	 * 插入论文
    	 */
    	consumePaperService.consumeZone(path,FilePool.ms_expert_paper,Gone.Mysql);
    	executorPool.lock();
    	
    	/**
    	 * 插入专利
    	 */
    	consumePatentsService.consumeZone(path,FilePool.ms_expert_patent,Gone.Mysql);
    	executorPool.lock();
    	
    	long e_ = System.currentTimeMillis();
		System.out.println("执行总时长:" + (float)(e_-s_)/1000);
    	
    	return success();
    }
	
    
    
}
